package com.liwy.produce.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Random;
import java.util.UUID;

/**
 * Copyright © 元气森林（北京）食品科技集团有限公司. All rights reserved.
 *
 * @Description:
 * @author: Crazy4J
 * @date: 2022年04月21日 14:55
 * @version: V1.0
 */
@RestController
@Slf4j
public class KafkaSendController {

    @Resource
    private KafkaTemplate<Object, Object> template;

    /**
     * ./kafka‐topics.sh ‐‐list ‐‐zookeeper localhost:2888
     * 购物
     */
    @GetMapping("/buy")
    public void buy() throws InterruptedException {
        for (int i = 0; i < 1000000; i++) {
            Thread.sleep(10);
            log.info("购物消息：" + i);
            String orderId = UUID.randomUUID().toString();
            //1.发送下单消息
            template.send("buy_topic", 0, "order_key", "order_msg_" + i + "_" + orderId);
            //2.发送付款消息
            template.send("buy_topic", 0, "pay_key", "pay_msg_" + i + "_" + orderId);
        }

    }


    @GetMapping("/send/{input}")
    public void sendFoo(@PathVariable String input) throws InterruptedException {

        for (int i = 0; i < 1; i++) {
            Thread.sleep(10L);
            String topic = "my-topic-test1";
            log.info("发送kafka信息, topic:" + topic);
            template.send(topic, input);
        }

//        ListenableFuture<SendResult<Object, Object>> send = template.send("topic-test1", input);
//
//        send.addCallback(o -> {
//            System.out.println("成功发送");
//        }, throwable -> {
//            System.out.println("发送失败");
//        });
    }

}
